【小ネタ】Lambda関数に画像データを渡す

山本紘暉

2021.06.04

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

（2021/6/6 内容を修正しました）

カフェチームの山本です。

現在、カフェのシステムでは、機械学習を用いて、カメラを用いて動画を撮影し、商品の前にいる人物の骨格や手を検出することで、どのユーザがどの商品を取り出したかを判定しています。

そのために、エッジデバイスからクラウドに動画を送信し、クラウド側で様々な処理を実行する、という構成を検討しており、最近の記事で、Kinesis Video Streamsから動画を取り出す方法、Lambdaレイヤを使ってTensorFlow Liteをインポートし、Lambda関数で推論処理を実行する方法ついて記載しました。

【Kinesis Video Streams】Lambda関数でKinesis Video Streamsから動画を取得する | DevelopersIO

Lambda関数でTensorFlow Liteをインポートし、推論処理を実行する | DevelopersIO

今回は、取り出した動画から画像を切り出して、推論処理を実行するLambda関数に画像データを渡す方法として、入力となるpayloadに直接入れ込む方法について記載します。他の方法としても、S3バケットに保存してキーを渡す・トリガーで実行する、SQSキューにメッセージを送信する、といった方法があります。なので、今回の方法は、他のリソースを用意するのが面倒だったり、他のサービスを介する時間が気になる場合に、一つの選択肢として考えてみてください。

実装コード

今回実装したコードは以下のとおりです。

入力方法

frameには画像データ（np.ndarray）を、lambda_function_nameには実行したいLambda関数の名前を入力する想定です（実行する際には、lambda_function_nameをinvokeする権限があるか、確認してください）。

import cv2
import base64
import json
import boto3

def invoke_infer_lambda_frame(frame, lambda_function_name):
    # compress image
    filepath_img = f"/tmp/img.png"
    cv2.imwrite(filepath_img, frame)
    img_bytes = open(filepath_img, "rb").read()

    # convert to b64-str
    img_b64 = base64.b64encode(img_bytes).decode("utf8")

    # invoke lambda
    data = {
        "img_b64": img_b64,
        "filename": "img.png",
    }
    payload = json.dumps(data)
    ret = boto3.client("lambda").invoke(
        FunctionName=lambda_function_name,
        InvocationType='RequestResponse', # Event(非同期) or RequestResponse(同期)
        Payload=payload,
    )

動作としては、以下の通りです

画像をファイルとして出力し、ファイルをバイナリとして読み込みなおす。これによってデータサイズを小さく（圧縮）します。（他にも良い方法があるかもしれません）
バイナリデータをbase64で変換した後、文字列に変換する。これによって、そのままpayloadに加えることはできないバイナリデータを、送信できる文字列に変換します。このとき、データサイズは、前のステップで圧縮した後の画像のサイズより、4/3倍のサイズに大きくなります。
payload用にjson形式で変換し、Lambda関数をinvokeする。invokeする際、画像のデータサイズによって、選べるInvocationTypeが異なります。Lambdaの制限として、payloadが256KB以下ならEvent（非同期）・RequestResponse（同期）の両方を選ぶことができ、256KBを超える場合はRequestResponse（同期）しか選べません（6MBを超える場合は、どちらも実行できません）。先程のbase64の変換で大きくなった後のサイズで決まるので注意が必要です。

受け取る方法

処理は以下のとおりです。先程の処理を逆順にたどる形です。

eventからパラメータを取り出す
文字列からバイナリに戻し、base64で画像データにもどす。
ファイルとして出力して読み直し、もとのframeに戻す。（今回は受け取ったfilenameをそのまま利用していますが、その必要はありませんが、圧縮時に利用したときと、同じ拡張子で出力する必要があります。）

import cv2
import base64

def lambda_handler(event, context):
    # get parameters
    img_b64 = event["img_b64"]
    filename = event["filename"]

    # convert to bytes
    img_bytes = base64.b64decode(img_b64.encode("utf8"))

    # decompress image
    with open("/tmp/" + filename, "wb") as file:
        file.write(img_bytes)
    img = cv2.imread(filepath_img)

    # ...